package com.chukun.tx.message.task;

import com.chukun.tx.constants.MessageConstant;
import com.chukun.tx.mapper.MqMessageMapper;
import com.chukun.tx.message.queue.DefaultMessageProcessor;
import com.chukun.tx.message.queue.MessageProcessor;
import com.chukun.tx.message.queue.TimeWheelQueueProcessor;
import com.chukun.tx.model.MqMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Component
public class MessageLoadTask {

    @Autowired
    private MqMessageMapper messageMapper;

    private ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1, r -> {
        Thread t = new Thread(r);
        t.setName("messageLoadTask thread");
        t.setDaemon(true);
        return t;
    }, new ThreadPoolExecutor.CallerRunsPolicy());


    @PostConstruct
    public void loadMessage() {
        executorService.scheduleAtFixedRate(() -> {
            List<Byte> statusList = Arrays.asList(MessageConstant.MessageStatus.MESSAGE_CREATED.getCode(),
                    MessageConstant.MessageStatus.MESSAGE_CONSUME_FAIL.getCode());
            List<MqMessage> mqMessages = messageMapper.selectMessagesByStatus(statusList, MessageConstant.TIME_INTERVAL.length);
            MessageProcessor defaultProcessor = DefaultMessageProcessor.getDefaultProcessor();
            for (MqMessage message : mqMessages) {
                try {
                    defaultProcessor.putMsg(message);
                } catch (Exception e) {
                }
            }
        }, 1, 5, TimeUnit.MINUTES);
    }
}
